/*
 * client_socket.cpp
 *
 *  Created on: 2021年1月27日
 *      Author: kimfeng
 */

#include "client_socket.h"
#include "buf_stream.hpp"
#include "logger.h"
#include <unistd.h>
#include <boost/bind/bind.hpp>

client_socket::client_socket(boost::asio::io_service &io_svc,
		const std::string &addr, const std::string &port) :
		m_magic(0), m_ver(0), m_seq(0),
		m_sock(io_svc),
		m_addr(addr),
		m_port(port),
		m_timer(io_svc),
		m_reconnect(0)
{
}

client_socket::~client_socket()
{
}

bool client_socket::connect(const std::string &address, const std::string &port)
{
	m_addr = address; m_port = port;
#if BOOST_ASIO_VERSION <= 101010
	boost::asio::ip::tcp::resolver resolver(m_sock.get_io_service());
	boost::asio::ip::tcp::resolver::query query(m_addr, m_port);
	boost::asio::ip::tcp::resolver::iterator itr = resolver.resolve(query);
	boost::asio::async_connect(m_sock, itr,
			boost::bind(&client_socket::handle_connect, this,
					boost::asio::placeholders::error));
#else
	boost::asio::ip::tcp::resolver resolver(m_sock.get_executor());
	boost::asio::ip::tcp::resolver::results_type endpoints =
			resolver.resolve(m_addr, m_port);
	boost::asio::async_connect(m_sock, endpoints.begin(), endpoints.end(),
			boost::bind(&client_socket::handle_connect, this,
	          boost::asio::placeholders::error));
#endif
	return true;
}

void client_socket::close()
{
	log_debug("connection %s:%u will be close",
			m_sock.remote_endpoint().address().to_string().c_str(),
			m_sock.remote_endpoint().port());
	m_sock.close();
}

void client_socket::handle_connect(const boost::system::error_code &e)
{
	if (!e)
	{
		log_debug("connect to server %s:%u", m_sock.remote_endpoint().address().to_string().c_str(),
				m_sock.remote_endpoint().port());
		on_connect();
		m_sock.async_read_some(boost::asio::buffer(m_buffer),
				boost::bind(&client_socket::handle_read, this,
						boost::asio::placeholders::error,
						boost::asio::placeholders::bytes_transferred));
	}
	else
	{
		log_error("connect to %s:%s: %s", m_addr.c_str(), m_port.c_str(), e.message().c_str());
		if (m_reconnect)
		{
			reconnect();
		}
	}
}

void client_socket::handle_read(const boost::system::error_code &e,
		std::size_t bytes_transferred)
{
	if (e)
	{
		log_debug("bytes_transferred %llu", bytes_transferred);
		log_error("errno %d message: %s", e.value(), e.message().c_str());
		log_debug("socket is %s open", m_sock.is_open() ? "still" : "not");
		close();
		if (m_reconnect) reconnect();
		return ;
	}
	m_timer.cancel();
	boost::tribool result;
	std::size_t bytes_parsed = 0;
	vec_data_t r;
	while (bytes_parsed != bytes_transferred)
	{
		boost::tie(result, boost::tuples::ignore) = m_parser.parse(m_pkg,
				m_payload, m_buffer.data() + bytes_parsed,
				m_buffer.data() + bytes_transferred, bytes_parsed);
		if (result)
		{
			vec_data_t data;
			sck_pkg_t pkg;
			data.swap(m_payload);
			m_pkg.payload = data.data();
			memcpy(&pkg, &m_pkg, sizeof(sck_pkg_t));
			memset(&m_pkg, 0, sizeof(sck_pkg_t));

			CallbackFunction f = m_handler.get_callback_func(pkg.req_num);
			if (NULL != f)
			{
				int ret = f(&pkg, r);
				if (0 != ret)
				{
					log_error("req_num %.4x callback function return %d", pkg.req_num, ret);
					close();
					return ;
				}
			}
			else
			{
				log_debug("not registered request num %.4x use on_read_rsp()", pkg.req_num);
				on_read_rsp(pkg);
			}
		}
		else if (!result)
		{
			log_error("Illegal socket package");
			close();
			return ;
		}
		else
		{
			m_timer.expires_from_now(boost::posix_time::seconds(3));
			m_timer.async_wait(boost::bind(&client_socket::handle_rd_timeout, this));
		}
	}

	m_sock.async_read_some(boost::asio::buffer(m_buffer),
			boost::bind(&client_socket::handle_read, this,
					boost::asio::placeholders::error,
					boost::asio::placeholders::bytes_transferred));
}

void client_socket::sck_pkg_new(sck_pkg_t *pkg, ushort req_num, ushort type,
		void *payload, std::size_t len)
{
	pkg->magic_id = m_magic;
	pkg->ver = m_ver;
	pkg->req_num = req_num;
	pkg->seq_num = ++m_seq;
	pkg->type = type;
	pkg->pending = 0;
	pkg->len = len;
	pkg->payload = payload;
}

size_t client_socket::sck_pkg_tobuf(const sck_pkg_t *pkg, uchar **buf)
{
	size_t n = 0;
	if (NULL == *buf)
	{
		*buf = (uchar *)malloc(pkg->len + PKG_HEADER_SIZE);
		if (NULL != *buf)
		{
			uchar *p = *buf;
			buf_write<ushort>(&p, pkg->magic_id);
			buf_write<ushort>(&p, pkg->ver);
			buf_write<ushort>(&p, pkg->req_num);
			buf_write<ushort>(&p, pkg->seq_num);
			buf_write<ushort>(&p, pkg->type);
			buf_write<ushort>(&p, pkg->pending);
			buf_write<uint>(&p, pkg->len);
			buf_writebytes(&p, (uchar *)pkg->payload, pkg->len);
			n = pkg->len + PKG_HEADER_SIZE;
		}
	}
	return n;
}

bool client_socket::send_buffer(uchar *buf, std::size_t n)
{
	m_sock.async_send(boost::asio::buffer(buf, n),
			boost::bind(&client_socket::handle_write, this,
					boost::asio::placeholders::error, buf));
	return true;
}

void client_socket::handle_rd_timeout()
{
	log_error("read socket package timeout at %s:%u",
			m_sock.remote_endpoint().address().to_string().c_str(),
			m_sock.remote_endpoint().port());
	close();
}

bool client_socket::send_pkg(const sck_pkg_t &pkg)
{
	uchar *buf = NULL;
	size_t n = sck_pkg_tobuf(&pkg, &buf);
	if (n == 0) return false;
	sck_pkg_print(pkg);
	return send_buffer(buf, n);
}

void client_socket::sck_pkg_print(const sck_pkg_t &pkg)
{
	fprintf(stdout, "pkg: Magic %.4x Ver %.4x Req %.4x Seq %.4X Len %u\n",
			pkg.magic_id, pkg.ver, pkg.req_num, pkg.seq_num, pkg.len);
}

void client_socket::handle_write(const boost::system::error_code &e, void *buf)
{
	if (NULL != buf) free(buf);
	if (e)
	{
		log_error("%s", e.message().c_str());
		close();
		if (m_reconnect) reconnect();
	}
}

bool client_socket::reconnect()
{
	sleep(m_reconnect);
	log_trace("try to reconnect %s:%s...", m_addr.c_str(), m_port.c_str());
#if BOOST_ASIO_VERSION <= 101010
	boost::asio::ip::tcp::resolver resolver(m_sock.get_io_service());
	boost::asio::ip::tcp::resolver::query query(m_addr, m_port);
	boost::asio::ip::tcp::resolver::iterator itr = resolver.resolve(query);
	boost::asio::async_connect(m_sock, itr,
			boost::bind(&client_socket::handle_connect, this,
					boost::asio::placeholders::error));
#else
	boost::asio::ip::tcp::resolver resolver(m_sock.get_executor());
	boost::asio::ip::tcp::resolver::results_type endpoints =
			resolver.resolve(m_addr, m_port);
	boost::asio::async_connect(m_sock, endpoints.begin(), endpoints.end(),
			boost::bind(&client_socket::handle_connect, this,
	          boost::asio::placeholders::error));
#endif
	return true;
}
